package io.renren.modules.job.task;

import io.renren.common.session.ClusterConfiguration;
import io.renren.common.session.SpringSessionConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;
import redis.clients.jedis.*;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.util.Pool;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by luzhenting on 2017/11/28.
 */
@Component("redisTask")
@PropertySource(value = { "classpath:/config.properties" })
public class RedisTask {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private static final String KEY_REDIS_NODES = "redis.nodes";

    /**
     * redis节点
     */
    @Value("${redis.nodes}")
    private String redisNodes;
    /**
     * 客户端连接重试次数
     */
    @Value("${redis.maxRedirects}")
    private String redisMaxRedirects;

    @Autowired
    private SpringSessionConnectionFactory connectionFactory;

    @Autowired
    private JedisPoolConfig poolConfig;

    @Autowired
    private ClusterConfiguration clusterConfiguration;

    /**
     * redis服务器列表是否产生变化的标识
     */
    private boolean isChange = false;

    /**
     * 客户端和rendis服务器定时连接，监测连接状态
     * 当某一个redis服务器宕机后，将其从服务器列表中踢除，并重新初始化redis连接池
     */
    public void run(){
        List<JedisShardInfo> availableNodes = new ArrayList<JedisShardInfo>();
        List<JedisShardInfo> nowNodes = clusterConfiguration.getShards();
        this.handleBedRedisNodes(nowNodes);
        for (JedisShardInfo shardInfo : nowNodes){
            Jedis jedis = shardInfo.createResource();
            for (int i=0; i<Integer.valueOf(redisMaxRedirects); i++){
                try {
                    if (jedis.ping().equals("PONG")) {
                        availableNodes.add(shardInfo);
                        break;
                    }
                } catch (JedisConnectionException je){
                    try {
                        //等待500毫秒
                        Thread.currentThread().sleep(500);
                    } catch (InterruptedException e) {
                        logger.error(e.getMessage(),e.getCause());
                    }
                }
                if (i == (Integer.valueOf(redisMaxRedirects)-1)){
                    isChange = true;
                    logger.warn("reids服务器host：",shardInfo.getHost(),"已下线，将从redis服务器列表踢除！");
                }
            }
        }
        if (isChange){
            isChange = false;
            //重新初始化redis连接池
            Pool<ShardedJedis> pool = new ShardedJedisPool(poolConfig,availableNodes);
            connectionFactory.setShardedJedisPool(pool);
            logger.info("redis连接池重新初始化结束！");
            //重置clusterConfiguration
            clusterConfiguration.setShards(availableNodes);
        }
    }

    /**
     * 处理损坏的redis服务器
     * 当redis服务器恢复之后重新添加到redis服务器列表中
     * @param nowNodes
     */
    private void handleBedRedisNodes(List<JedisShardInfo> nowNodes){
        List<JedisShardInfo> allNodes = ClusterConfiguration.converStringToSet(redisNodes);
        List<JedisShardInfo> copy = new ArrayList<JedisShardInfo>(nowNodes);
        for (JedisShardInfo node : allNodes){
            int i = 0;
            for (JedisShardInfo nowNode : copy){
                i++;
                if (node.getHost().equals(nowNode.getHost())
                        && node.getPort()==nowNode.getPort()){
                    break;
                }else if (i == nowNodes.size()){
                    Jedis jedis = node.createResource();
                    try {
                        if (jedis.ping().equals("PONG")) {
                            nowNodes.add(node);
                            logger.info("reids服务器"+node.getHost()+":"+node.getPort()+"单次尝试恢复成功");
                        }
                    } catch (JedisConnectionException e){
                        logger.warn("reids服务器"+node.getHost()+":"+node.getPort()+"单次尝试恢复失败");
                    }
                }
            }
        }
    }
}
